package com.bindatax.event.impl.statistics.dblog;

import com.bindatax.core.custom.StatusCode;
import com.bindatax.core.log.LogOutManager;
import com.bindatax.core.meta.TaskMeta;
import com.bindatax.core.model.ValueMetaData;
import com.bindatax.core.statistics.StatisticsData;
import com.bindatax.core.task.DataTask;
import com.bindatax.core.task.event.BaseListenEvent;
import com.bindatax.core.task.event.EventScope;
import com.bindatax.core.task.event.EventType;
import com.bindatax.core.task.event.ListenEventInterface;
import com.bindatax.core.task.row.EventValueData;
import com.bindatax.core.task.segment.SegmentWork;
import com.bindatax.core.util.BinDataxStringUtil;
import com.bindatax.core.util.BinDataxTimeUtil;
import com.bindatax.event.impl.statistics.dblog.data.LogDataOperate;
import com.bindatax.event.impl.statistics.dblog.model.RunTimeLog;
import com.bindatax.event.impl.statistics.dblog.model.ZlLog;
import org.pentaho.di.core.row.ValueMetaInterface;

import java.sql.Timestamp;
import java.util.HashSet;
import java.util.Properties;

public class TaskEventStatisticsSaveLog extends BaseListenEvent implements ListenEventInterface {
    private LogDataOperate logDataOperate;
    private RunTimeLog runTimeLog;
    private ZlLog zlLog;
    private boolean checkDataMaxTime = false;
    private Timestamp dataMaxTime = new Timestamp(0L);
    private String queryIncrementTimeField;//源表增量时间字段

    private TaskMeta taskMeta;
    private Properties taskPros;

    @Override
    public int getOrder() {
        return 0;
    }

    @Override
    public EventScope getScope() {
        return EventScope.PROTOTYPE;//多例
    }

    @Override
    public HashSet<EventType> getTypes() {
        eventTypes.add(EventType.TASK_RUNNING_CREATE_CONN_AFTER);//构建连接池之后
        eventTypes.add(EventType.TASK_RUNNING_SOURCE_QUERY_BEFORE);//查询结果之前
        eventTypes.add(EventType.TASK_RUNNING_QUERY_VALUE_DATA);//查询结果之前
        eventTypes.add(EventType.TASK_FINISH_SUCCESS);//查询结果之前
        eventTypes.add(EventType.TASK_FINISH_FAIL);//查询结果之前
        return eventTypes;
    }

    @Override
    public void exe(EventType type, DataTask thisTask, SegmentWork thisSegmentWork, EventValueData rowData) {
        super.exe(type,thisTask,thisSegmentWork,rowData);
        if(stop){
            LOG.error("执行事件错误（事件实例已经停止！）：" + type.getDescription());
            return;
        }

        // 初始化数据
        if(EventType.TASK_RUNNING_CREATE_CONN_AFTER == type && logDataOperate == null){
            LOG.debug("初始化事件数据：" + type.getDescription());
            runTimeLog = new RunTimeLog();
            taskMeta = thisTask.getTaskMeta();
            taskPros = taskMeta.getPros();

            runTimeLog.setXzqhdm( taskMeta.getBizValue("xzqhdm") );
            runTimeLog.setXzqhmc( taskMeta.getBizValue("xzqhmc") );
            runTimeLog.setSchema(taskMeta.getQuerySchema());
            runTimeLog.setTableName(taskMeta.getTargetTableName());
            runTimeLog.setStatusCode(StatusCode.SUCCESS);
            String sqlId = BinDataxStringUtil.md5(taskMeta.getQuerySql());

            String sqlGuid = taskMeta.getBizValue("sql_guid");

            runTimeLog.setSqlGuid(sqlGuid);
            runTimeLog.setStartTime(new Timestamp(System.currentTimeMillis()));
            runTimeLog.setGroupBy( taskMeta.getBizValue("group_by") );
            runTimeLog.setRemarks( taskMeta.getBizValue("remarks") );

            zlLog = new ZlLog();
            String id = ZlLog.createId( runTimeLog.getXzqhdm() , runTimeLog.getSchema() ,sqlId);
            zlLog.setId(id);
            zlLog.setXzqhdm(runTimeLog.getXzqhdm());
            zlLog.setXzqhmc(runTimeLog.getXzqhmc());
            zlLog.setSqlGuid(sqlGuid);
            zlLog.setSchema(runTimeLog.getSchema());
            zlLog.setTableName(runTimeLog.getTableName());

            logDataOperate = new LogDataOperate(taskMeta);
            LOG.debug("事件数据初始化完成：" + type.getDescription());
            return;
        }
        if(logDataOperate == null){
            LOG.error("执行事件错误（事件实例未初始化！）：" + type.getDescription());
            return;
        }

        // 查询增量时间
        if(EventType.TASK_RUNNING_SOURCE_QUERY_BEFORE == type){
            queryIncrementTimeField = taskMeta.getQueryIncrementTimeField();
            checkDataMaxTime = queryIncrementTimeField != null && queryIncrementTimeField.length() > 0;
            if (checkDataMaxTime){//如果存在时间字段，则从数据库获取增量时间
                String sqlId = BinDataxStringUtil.md5(taskMeta.getQuerySql());
                String id = ZlLog.createId( runTimeLog.getXzqhdm() , runTimeLog.getSchema() ,sqlId);
                Timestamp zlsj = logDataOperate.queryZlTime(id);
                //如果为空，则直接设置为最小值
                if(zlsj == null){
                    zlsj = new Timestamp(0L);
                }
                runTimeLog.setZlOldTime(zlsj);
                dataMaxTime = zlsj;

                //增量时间设置，看看SQL语句有没有时间字段替换
                String runTimeSql = taskMeta.getQuerySql();
                //进行判断，看看是否需要增量获取数据
                if(runTimeSql.contains("#QueryTime#")){
                    Timestamp queryTime = new Timestamp( zlsj.getTime() + 1000);//加多一秒，防止精确度不够导致获取到旧数据。
                    String time = BinDataxTimeUtil.timestampFormatString(queryTime);
                    LOG.debug("本次增量开始时间为(已自动加多1秒，防止比较时间精度不够，获取到旧数据。)：" + time);
                    runTimeSql = runTimeSql.replaceAll("#QueryTime#", " TO_TIMESTAMP('" + time + "', 'yyyy-mm-dd hh24:mi:ss.FF3') ");
                    thisTask.setRunTimeSql(runTimeSql);
                }else {
                    LOG.warn("查询SQL里没有增量时间查询条件占位符：#QueryTime# ，不加入时间条件查询。" );
                }
                LOG.info("实时查询SQL语句为："+ LogOutManager.CR + runTimeSql);
            }
            return;
        }

        // 获取最大时间
        if(EventType.TASK_RUNNING_QUERY_VALUE_DATA == type && checkDataMaxTime && rowData.isState() ){
            ValueMetaData metaData = rowData.getMetaData();
            ValueMetaInterface commValueMeta = metaData.getValueMetaInterface();
            String resColName = commValueMeta.getName();
            // 在这里获取增量时间字段
            if (resColName.equalsIgnoreCase(queryIncrementTimeField)) {
                Object time = rowData.getValue();
                if(time != null){
                    try {
                        Timestamp tmp = Timestamp.valueOf(time.toString());
                        if(tmp.after(dataMaxTime)){
                            dataMaxTime = tmp;
                        }
                    } catch (Exception e) {
                        LOG.error(String.format("获取增量时间字段值[%s]发生错误:%s",queryIncrementTimeField, e.getMessage()) , e);
                    }
                }
            }
            return;
        }

        if(EventType.TASK_FINISH_SUCCESS == type || EventType.TASK_FINISH_FAIL == type){
            Timestamp endTime = new Timestamp(System.currentTimeMillis());
            zlLog.setZlsj(dataMaxTime);
            runTimeLog.setZlNewTime(dataMaxTime);
            runTimeLog.setEndTime(endTime);
            if(EventType.TASK_FINISH_FAIL == type){
                runTimeLog.setStatusCode(StatusCode.FAIL);
                Exception taskError = thisTask.getTaskError();
                if(taskError != null){
                    runTimeLog.setErrorInfo("任务执行失败",taskError);
                }
            }
            StatisticsData statisticsData = thisTask.getTaskStatisticsData();
            // 写入统计数据
            runTimeLog.readStatisticsData(statisticsData.getTaskStatistics());
            // 写入数据
            logDataOperate.saveRunTimeLog(runTimeLog);
            logDataOperate.saveZlTime(zlLog);
            stop = true;
        }

    }

}
